package com.wugui.datatx.core.thread;

import java.util.concurrent.TimeUnit;

import com.wugui.datatx.core.biz.AdminBiz;
import com.wugui.datatx.core.biz.model.RegistryParam;
import com.wugui.datatx.core.biz.model.ReturnT;
import com.wugui.datatx.core.enums.RegistryConfig;
import com.wugui.datatx.core.executor.JobExecutor;
import com.wugui.datatx.core.util.OSUtils;

import lombok.extern.slf4j.Slf4j;

/**
 * Created by xuxueli on 17/3/2.
 */
@Slf4j
public class ExecutorRegistryThread {

	private static ExecutorRegistryThread instance = new ExecutorRegistryThread();

	public static ExecutorRegistryThread getInstance() {
		return instance;
	}

	private Thread registryThread;
	private volatile boolean toStop = false;

	public void start(final String appName, final String address) {

		// valid
		if (appName == null || appName.trim().length() == 0) {
			log.warn(">>>>>>>>>>> datax-web, executor registry config fail, appName is null.");
			return;
		}
		if (JobExecutor.getAdminBizList() == null) {
			log.warn(">>>>>>>>>>> datax-web, executor registry config fail, adminAddresses is null.");
			return;
		}

        log.info(">>>>>>>>>>> datax-web, 执行器注册 starting ......");

		registryThread = new Thread(() -> {

			// registry
			while (!toStop) {
				try {
					RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appName,
							address, OSUtils.cpuUsage(), OSUtils.memoryUsage(), OSUtils.loadAverage());
					for (AdminBiz adminBiz : JobExecutor.getAdminBizList()) {
						try {
							ReturnT<String> registryResult = adminBiz.registry(registryParam);
							if (registryResult != null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {
								registryResult = ReturnT.SUCCESS;
								log.debug(
                                    ">>>>>>>>>>> datax-web 执行器注册 success, registryParam:{}, registryResult:{}",
										new Object[] { registryParam, registryResult });
								break;
							} else {
								log.info(
                                    ">>>>>>>>>>> datax-web 执行器注册 fail, registryParam:{}, registryResult:{}",
										new Object[] { registryParam, registryResult });
							}
						} catch (Exception e) {
                            log.info(">>>>>>>>>>> datax-web 执行器注册 error, registryParam:{}", registryParam, e);
						}

					}
				} catch (Exception e) {
					if (!toStop) {
						log.error(e.getMessage(), e);
					}

				}

				try {
					if (!toStop) {
						TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
					}
				} catch (InterruptedException e) {
					if (!toStop) {
						log.warn(
                            ">>>>>>>>>>> datax-web, 执行器注册 thread interrupted, error msg:{}",
								e.getMessage());
					}
				}
			}

			// registry remove
			try {
				RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appName,
						address);
				for (AdminBiz adminBiz : JobExecutor.getAdminBizList()) {
					try {
						ReturnT<String> registryResult = adminBiz.registryRemove(registryParam);
						if (registryResult != null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {
							registryResult = ReturnT.SUCCESS;
							log.info(
									">>>>>>>>>>> datax-web registry-remove success, registryParam:{}, registryResult:{}",
									new Object[] { registryParam, registryResult });
							break;
						} else {
							log.info(
									">>>>>>>>>>> datax-web registry-remove fail, registryParam:{}, registryResult:{}",
									new Object[] { registryParam, registryResult });
						}
					} catch (Exception e) {
						if (!toStop) {
							log.info(">>>>>>>>>>> datax-web registry-remove error, registryParam:{}",
									registryParam,
									e);
						}

					}

				}
			} catch (Exception e) {
				if (!toStop) {
					log.error(e.getMessage(), e);
				}
			}
			log.info(">>>>>>>>>>> datax-web, executor registry thread destory.");

		});
		registryThread.setDaemon(true);
		registryThread.setName("datax-web, executor ExecutorRegistryThread");
		registryThread.start();
	}

	public void toStop() {
		toStop = true;
		// interrupt and wait
		registryThread.interrupt();
		try {
			registryThread.join();
		} catch (InterruptedException e) {
			log.error(e.getMessage(), e);
		}
	}

}